/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package aiaudio.processing.ratings.normilize.zscore.reducers;

import aiaudio.processing.base.mapreduce.CoordinateDoubleValueKey;
import aiaudio.Application;
import aiaudio.database.tablegroup.InitialDataTableGroup;
import aiaudio.database.tables.DataMetricsTable;
import aiaudio.database.tables.RatingsMatrixTable;
import aiaudio.database.tables.UsersListTable;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableReducer;

/**
 *
 * @author nastya
 */
public abstract class ZScoreUserToArtistValueReducer extends TableReducer<ImmutableBytesWritable, CoordinateDoubleValueKey, ImmutableBytesWritable> {

    @Override
    protected void reduce(ImmutableBytesWritable key, Iterable<CoordinateDoubleValueKey> values, Context context) throws IOException, InterruptedException {
        long totalListenCountByUser = getTotalListenCountByUser(key.get());
        long totalVotingByUser = getTotalVotingByUser(key.get());
        long totalArtistsCount = getTotalArtistsCount();
        double defaultVoting = calculateDefaultVoting(totalListenCountByUser, totalVotingByUser);
        double meanValue = (double) totalListenCountByUser / (double) totalVotingByUser;
        double sum = 0;
        List<CoordinateDoubleValueKey> lst = new ArrayList<CoordinateDoubleValueKey>();
        double maxDeviation = 0;
        for (CoordinateDoubleValueKey artistValueKey : values) {
            lst.add(new CoordinateDoubleValueKey(artistValueKey.getValue(), artistValueKey.getCoordinate()));
            double dev = artistValueKey.getValue() - meanValue;
            if (Math.abs(dev) > maxDeviation) {
                maxDeviation = Math.abs(dev);
            }
            sum += (dev) * (dev);
        }
        sum += (totalArtistsCount - totalVotingByUser) * (defaultVoting - meanValue) * (defaultVoting - meanValue);
        double standardDeviation = Math.sqrt(sum / totalVotingByUser);
        maxDeviation = Math.floor(maxDeviation);
        Put put = new Put(key.get());

        for (CoordinateDoubleValueKey artistValueKey : lst) {
            double result = (artistValueKey.getValue() - meanValue) / standardDeviation;
            result = (result + maxDeviation) / (2 * maxDeviation);
            RatingsMatrixTable.addUserRating(put, artistValueKey.getCoordinate() ,result);
        }
        double defaultVotingProcessed = (defaultVoting - meanValue) / standardDeviation;
        RatingsMatrixTable.addDefaultRating(put, defaultVotingProcessed);
        context.write(key, put);
    }

    private long getTotalListenCountByUser(byte[] user) throws IOException {
        HTable usersTable = Application.database().getByName(InitialDataTableGroup.getInstance().getUsersTable()).getTable();
        return UsersListTable.getListenCount(usersTable, user);
    }

    private long getTotalVotingByUser(byte[] user) throws IOException {
        HTable usersTable = Application.database().getByName(InitialDataTableGroup.getInstance().getUsersTable()).getTable();
        return UsersListTable.getVotingCount(usersTable, user);
    }

    private long getTotalArtistsCount() throws IOException {
        HTable dataMetricsTable = Application.database().getByName(InitialDataTableGroup.getInstance().getDataMetricsTable()).getTable();
        return DataMetricsTable.getArtistsCount(dataMetricsTable);
    }

    protected abstract double calculateDefaultVoting(long totalListenCountByUser, long totalVotingByUser);
}
